pytest-xdistでPytestの並列実行をしてみる

pytest-xdistでPytestの並列実行をしてみる

#Python
#pytest
#単体テスト
#smallテスト
kobayashi.m

kobayashi.m

facebook logohatena logotwitter logo
Clock Icon2024.02.26

はじめに

データアナリティクス事業本部のkobayashiです。

Pytestでテストするsmallテストが多数ある場合はテストを並列実行してテスト結果を早く取得することが開発スピードを上げる一つの要素になるかと思います。今回はPytestでテストを並列実行してみたのでその内容をまとめます。

Pytestを並列実行するプラグイン

Smallテストは単一プロセス内で動作するテストなので非常に高速に実行されかつスケールさせる事ができます。そのためpytestで並列実行させるには最適なテストとなります。

pytestで並列テストを行うにはpytestのプラグインを使いますが、調べてみると以下の2つが候補に上がりました。

それぞれのドキュメントやリリース履歴を見た所pytest-parallelに関してはここ数年リリースがなく開発が止まっていると思われるのでpytest-xdistを今後は使用すべきかと思い今回もpytest-xdistを使っていきたいと思います。

pytest-xdistを使ってみる

環境

  • Python: 3.11.4
  • pytest: 7.4.3

インストールはいつも通りpipでインストールします。

$ pip install pytest-xdist
$  pip list | grep xdist
pytest-xdist                     3.5.0

テスト対象とテストコード

テスト対象のコードは以下のモンテカルロ法で円周率を求める関数になります。

import numpy as np


def calc_pi(sim_num):
    count = 0

    for i in range(sim_num):

        x = np.random.rand()
        y = np.random.rand()

        d = x ** 2 + y ** 2

        if d <= 1:
            count += 1

    p = count / sim_num * 4

    return p

import pytest

from main import calc_pi


class TestPi_1:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_1(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_2(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14


class TestPi_2:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_2(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_4(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14


import pytest

from main import calc_pi


class TestPi_3:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_5(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_6(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14


class TestPi_4:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_7(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_8(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

ではこれらのコードでテストを並列で行ってみます。

並列実行でテストする

python-xdistを使って並列でテストを行うには-n autoのオプションを付けてpytestを実行するだけです。

$ pytest -v -n auto  
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw3] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw4] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
[gw1] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
[gw7] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw5] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw6] [ 87%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw2] [100%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]

[gw0]等がpytest-xdistのワーカー名になります。

-n autoで実行するとpytest-xdistが実行環境マシーンのCPUのコアと同じ数のワーカー数でテストが並列実行されます。上記のコードだと8つのワーカーで8つのテストを行っているため大幅なテスト時間の短縮になります。

ワーカー数を指定する場合は-n {ワーカー数}でワーカー数を指定することで指定のワーカー数でテストを並列実行できます。

$ pytest -v -n 2 
=========================================================================== test session starts ============================================================================
2 workers [8 items]     
scheduling tests via LoadScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
[gw1] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw0] [ 25%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw1] [ 37%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw1] [ 62%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 87%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]

このような感じでワーカーが2つでテストが並列実行されています。

テスト配布方法（クラス単位）

pytest-xdistで特にオプションを指定しない場合はテストをどのワーカーで動かすかは自動的に利用可能なワーカーに割り振られ実行されます。したがってここまでの例ではクラスごと、ファイルごとにテストを記述してありますがテストの実行順序が担保されません。（--dist loadオプションがデフォルトなため）

そこでpytest-xdistでは配布オプションを使うことでクラス、ファイル、グループ単位で同一のワーカー上でテストを行うことができるので試してみます。

はじめにクラス単位でテストを並列実行してみます。この場合は--dist loadscopeでテストを実行します。

$ pytest -v -n auto --dist loadscope    
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadScopeScheduling

test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw3] [ 12%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw2] [ 50%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw3] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 87%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw2] [100%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]

ワーカー数はautoなので8つ立ち上がっていますがgw0のワーカーでTestPi_1クラス、gw1のワーカーでTestPi_3クラス、gw2のワーカーでTestPi_2クラス、gw3のワーカーでTestPi_4クラスとテストコードのクラスが4つなので、4つのワーカーでクラスが1つづつ実行されていることがわかります。

テスト配布方法（ファイル）

次にファイル単位位でテストを並列実行してみます。この場合は--dist loadfileでテストを実行します。

 pytest -v -n auto --dist loadfile
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadFileScheduling

test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_1::test_calc_pi_1[10000000] 
[gw0] [ 12%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw1] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
[gw1] [ 50%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [ 62%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw0] [ 87%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw1] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]

こちらもワーカーは8つ立ち上がっていますがgw0のワーカーでtest_main.pyファイル、gw1のワーカーでtest_main_2.pyファイルとテストコードのファイルが2つなので、2つのワーカーでファイル単位で1つづつ実行されていることがわかります。

テスト配布方法（指定グループ）

次に指定したグループ単位でテストを並列実行してみます。この場合は--dist loadgroupでテストを実行しますが、その前にグループを分ける必要があります。先ほど作成したtest_main.py、test_main_2.pyのクラスに@pytest.mark.xdist_groupデコレータを使ってグループ分けをしてみます。

import pytest

from main import calc_pi

@pytest.mark.xdist_group(name="GroupA")
class TestPi_1:
...


@pytest.mark.xdist_group(name="GroupB")
class TestPi_2:
...


import pytest

from main import calc_pi


@pytest.mark.xdist_group(name="GroupA")
class TestPi_3:
...


@pytest.mark.xdist_group(name="GroupB")
class TestPi_4:
...


$ pytest -v -n auto --dist loadgroup
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadGroupScheduling

test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupB 
test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
[gw0] [ 12%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
[gw1] [ 25%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupB 
test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
[gw1] [ 50%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
[gw0] [ 62%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 87%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw1] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB

こちらもワーカーは8つ立ち上がっていますがグループを2つ作成したのでgw0のワーカーでGroupAのグループを付けたTestPi_1・TestPi_3クラス、gw1のワーカーでGroupBのデコレータを付けたTestPi_2・TestPi_4クラスとテストコードのグループが2つなので、2つのワーカーで指定したグループ単位に実行されていることがわかります。

試しに以下のようにクラスではなくメソッドでグループを指定することもできます。

import pytest

from main import calc_pi

@pytest.mark.xdist_group(name="GroupA")
class TestPi_1:
...


class TestPi_2:
    @pytest.mark.xdist_group(name="GroupA")
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_3(self, data_in):
...

    @pytest.mark.xdist_group(name="GroupB")
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_4(self, data_in):
...

この場合は以下のような実行結果になります。

$ pytest -v -n auto --dist loadgroup
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadGroupScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
[gw1] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
[gw0] [ 25%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
[gw1] [ 37%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupA 
[gw1] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 75%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
[gw0] [ 87%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw0] [100%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA

まとめ

Pytestでテストを並列実行するpytest-xdistプラグインを使ってみました。外部に依存関係が無く1プロセス上で動くようなテストでしたら-n autoで自動分散させてしまえば効率的にテストを行えますし、ある程度順序を制御したい場合は--distで適切な分散を指定すればいいので簡単に使えるかと思います。

最後まで読んで頂いてありがとうございました。

Share this article

facebook logohatena logotwitter logo

関連記事

[アップデート] AWS Lambda で Python 3.13 ランタイムがサポートされました

[アップデート] AWS Lambda で Python 3.13 ランタイムがサポートされました

User avatar
いわさ
2024.11.14
OpenPyXLを使ってPythonでエクセルファイルに画像を追加する

OpenPyXLを使ってPythonでエクセルファイルに画像を追加する

User avatar
藤井元貴
2024.11.12
VSCodeでJupyter Notebookに依存しないPython Native REPLを使ってみた

VSCodeでJupyter Notebookに依存しないPython Native REPLを使ってみた

User avatar
なかた
2024.11.12
Python3.12 のスタンドアロン Lambda 関数を作成しつつ、SAM CLIを学んでみた

Python3.12 のスタンドアロン Lambda 関数を作成しつつ、SAM CLIを学んでみた

User avatar
なかた
2024.11.08
Classmethod's white logo
Facebook's logo
X's logo
YouTube's logo

© Classmethod, Inc. All rights reserved.